package org.apache.flink.streaming.api.datastream;

import iscas.RedisConfig;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import redis.clients.jedis.Jedis;

public class ABSProcessFunction<K, I extends ABSMarkable, O> extends KeyedProcessFunction<K, I, O> {
    Jedis jedis;

    @Override
    public void open(Configuration parameters) {
        jedis = RedisConfig.newJedis();
    }

    private void incrChannelCount() {
        jedis.incr("ABSCount");
    }

    @Override
    public void processElement(I value, Context ctx, Collector<O> out) {
        if (value.hasMark()) {
            incrChannelCount();
        }
    }
}
